package com.shujia.stream

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

import scala.collection.mutable.ListBuffer

object Demo4Bukong {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("stream")

    val sc = new SparkContext(conf)

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    //布控列表


    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)


    /**
      * transform 将ds 转换成rdd , 每一个batch都会执行一次
      * transform 里面还是属于Driver端
      *
      */


    val filterDS: DStream[String] = linesDS.transform(rdd => {

      //从数据库中查询布控列表
      println("获取布控列表")
      val mdns: List[String] = getList

      //获取sparkContext
      val context: SparkContext = rdd.sparkContext

      /**
        * 动态修改广播变量，每隔5秒广播一次
        *
        */

      //广播
      val bro: Broadcast[List[String]] = context.broadcast(mdns)


      //过滤数据
      val filterRDD: RDD[String] = rdd.filter(line => {
        val split: Array[String] = line.split(",")
        val mdn: String = split(0)
        bro.value.contains(mdn)
      })


      //返回一个rdd
      filterRDD
    })

    /**
      * foreachRDD ： 将DS转换成rdd  没有返回值
      *
      */


    //保存数据到mysql
    filterDS.foreachRDD(rdd => {


      rdd.foreachPartition(iter => {


        /**
          * foreachPartition : 只为每一个分区创建一个链接
          *
          */

        //查询数据库获取布控列表
        Class.forName("com.mysql.jdbc.Driver")

        val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")


        //循环一个分区的数据，将数据保存到mysql中
        iter.foreach(line => {

          val stat: PreparedStatement = con.prepareStatement("insert into t_result(line) values(?)")

          stat.setString(1, line)

          stat.executeUpdate()

        })

        con.close()
      })

    })


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }

  /**
    * 获取布控列表
    *
    * @return
    */

  def getList: List[String] = {
    val mdns = new ListBuffer[String]

    //查询数据库获取布控列表
    Class.forName("com.mysql.jdbc.Driver")

    val con: Connection = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")

    val stat: PreparedStatement = con.prepareStatement("select mdn from t_mdns")

    val resultSet: ResultSet = stat.executeQuery()

    while (resultSet.next()) {
      val mdn: String = resultSet.getString("mdn")

      mdns += mdn
    }

    con.close()

    //返回数据

    mdns.toList
  }

}
